package com.yujiahao.bigdata.rdd.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * TODO 1) 某些场合需要，降低（减少）分区数量，采用特殊的算子
 * 1、coalesce算子可以缩减分区，目的是为了减少分区的调度以及资源的合理利用
 * 2、在进行分区的时候，默认不会使用shuffle操作，
 * 3、但是可能会导致数据不均衡，所以采用shuffle让数据打乱重新组合的，让数据均衡一些。
 * 4、第一个参数表示修改分区的数量
 * 5、第二个参数表示是否使用shuffle，默认为false不洗牌，我们设置成true
 * 6、默认是么有shuffle的，因此不会打乱组合，如果不采用shuffle操作，那么coalesce算子不会增大分区
 * 7、shuffle会将数据打乱重组，但是不一定让数据均衡
 * val value: RDD[Int] = rdd.coalesce(4,true)
 * TODO 2）Spark专门提供了一个专门改变分区的算子，所以coalesce一般要用于【缩减】分区
 * 扩大分区一定是要shuffle的，repatition方法的底层还是通过coalesce实现的
 */

object Spark_RDD_Pper_Transform_Coalesce {
  def main(args: Array[String]): Unit = {


    //TODO 1、获得连接
    val conf: SparkConf = new SparkConf().setAppName("WordCount").setMaster("local")
    val sc: SparkContext = new SparkContext(conf)

    //TODO 3、 业务逻辑--测试coalesce[缩减分区--默认没有]、repatition[扩大分区--有shuffle]
    val rdd = sc.makeRDD(List(1, 2, 3, 4, 5, 6), 3)

    //测试初始分区
    rdd.saveAsTextFile("output")
    //测试扩大分区
    val value: RDD[Int] = rdd.repartition(6)
    value.saveAsTextFile("output1")
    //测试缩小分区
    val newRDD: RDD[Int] = value.coalesce(1)
    newRDD.saveAsTextFile("output2")
    // value.collect().foreach(println)

    //TODO 2、关闭连接
    sc.stop()
  }
}